package edu.zju.gis.dbfg.model.loader;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;

import java.io.IOException;
import java.io.Serializable;
import java.lang.reflect.Field;
import java.lang.reflect.ParameterizedType;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * @author Hu
 * @date 2019/10/7
 **/
public abstract class DataLoader<T> implements Serializable {

  protected abstract void init(String[] args);

  protected abstract boolean run(JavaSparkContext jsc) throws Exception;

  protected Class<T> getTClass() {
    return (Class <T>) ((ParameterizedType) getClass().getGenericSuperclass()).getActualTypeArguments()[0];
  }

  protected List<Field> getFields(){
    List<Field> fields = new ArrayList<>();
    Class <T> c = getTClass();
    // 迭代获取类及父类中的所有字段
    while(c!=null){
      fields.addAll(Arrays.asList(c.getDeclaredFields()));
      c = (Class <T>)c.getSuperclass();
    }
    return fields;
  }

  protected abstract ESDataLoader2<T>.IngestResult ingest(JavaRDD<T> featuresRDD, String sourceName) throws Exception;

  protected abstract void calExtents(JavaRDD<T> featuresRDD);

  protected abstract void finish() throws IOException;

}
